/*
Copyright (C) 2008  Helmut Juskewycz
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package com.juskewycz.dividere.action.split;

import com.juskewycz.dividere.action.ProgressStatus;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Implements <code>ISplitter</code> and uses multiple threads for the splitting.
 * 
 * @author Helmut Juskewycz
 * @createdon 27. January 2008
 * @lastmodified 27. January 2008
 * @history
 * 27. January 2008 Finished version 1.0
 */
public class MultiThreadBufferSplitter extends ASplitter {

    private static final Logger logger = Logger.getLogger(MultiThreadBufferSplitter.class.getName());
    private static final int THREAD_POOL_SIZE = 4;
    private int piecesNumber;
    private int piecesProcessed;
    private boolean errorOccured;

    @Override
    public void split(String filePath, long splitSize) {

	if (!validSplitterInputArgumtents(filePath, splitSize)) {
	    return;
	}

	File originalFile = new File(filePath);

	if (!checkSplitFileExists(originalFile)) {
	    return;
	}

	long wholeFileSize = originalFile.length();
	piecesNumber = (int) (wholeFileSize / splitSize + 1);
	final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

	try {

	    notifySplitBegin(originalFile);
	    notifyProgressListeners(new ProgressStatus(0, piecesNumber, ""));

	    piecesProcessed = 0;
	    int piecesQueued = 0;

	    errorOccured = false;

	    while (piecesQueued < piecesNumber) {
		executorService.execute(new SplitRunnable(originalFile, piecesQueued, piecesNumber, splitSize));
		piecesQueued++;
	    }
	    executorService.shutdown();
	    executorService.awaitTermination(300, TimeUnit.SECONDS);
	    if (!errorOccured) {
		notifySplitEnd(originalFile);
		notifyProgressListeners(new ProgressStatus(piecesNumber, piecesNumber, ""));
	    }
	} catch (InterruptedException ex) {
	    logger.log(Level.SEVERE, null, ex);
	    notifyError(new ProgressStatus(piecesProcessed, piecesNumber, ERROR_FILE_IO, true));
	}
    }

    private synchronized void setErrorOccured(boolean error) {
	errorOccured = error;
    }

    private synchronized void pieceProcessed(File filePiece, int pieceNumber) {
	piecesProcessed++;
	notifySplitPartEnd(pieceNumber, filePiece);
	notifyProgressListeners(new ProgressStatus(piecesProcessed, piecesNumber, filePiece.getPath()));
    }

    private class SplitRunnable implements Runnable {

	private byte[] buffer;
	private File wholeFile;
	private final int pieceNumber;
	private final int piecesNumber;
	private final long splitSize;

	public SplitRunnable(File wholeFile, int pieceNumber, int piecesNumber, long splitSize) {
	    this.wholeFile = wholeFile;
	    this.pieceNumber = pieceNumber;
	    this.piecesNumber = piecesNumber;
	    this.splitSize = splitSize;
	}

	public void run() {
	    buffer = new byte[BUFF_SIZE];
	    File filePiece = null;
	    InputStream in = null;
	    OutputStream out = null;
	    try {
		in = new FileInputStream(wholeFile);
		filePiece = new File(createSplitFilePath(wholeFile.getPath(), pieceNumber + 1));
		long offset = splitSize * pieceNumber;
		notifySplitPartBegin(pieceNumber, filePiece);

		if (filePiece.createNewFile()) {
		    out = new FileOutputStream(filePiece);
		    in.skip(offset);
		    int readBytes = 0;
		    while (!errorOccured && (readBytes + BUFF_SIZE) < splitSize) {
			int amountRead = in.read(buffer);
			if (amountRead == -1) {
			    break;
			}
			readBytes += amountRead;
			out.write(buffer, 0, amountRead);
		    }

		    byte[] restBuffer = new byte[(int) (splitSize - readBytes)];
		    int amountRead = in.read(restBuffer);
		    if (amountRead != -1) {
			out.write(restBuffer, 0, amountRead);
		    }
		    restBuffer = null;
		    pieceProcessed(filePiece, pieceNumber);
		} else {
		    setErrorOccured(true);
		    errorCanNotCreatePartFile(filePiece.getPath(), piecesNumber, pieceNumber);
		}
	    } catch (IOException e) {
		if (filePiece != null) {
		    filePiece.delete();
		}
		logger.log(Level.SEVERE, null, e);
		notifyError(new ProgressStatus(piecesProcessed, piecesNumber, ERROR_FILE_IO, true));
	    } finally {
		if (in != null) {
		    try {
			in.close();
		    } catch (IOException ex) {
			Logger.getLogger(MultiThreadBufferSplitter.class.getName()).log(Level.SEVERE, null, ex);
		    }
		}
		in = null;

		if (out != null) {
		    try {
			out.flush();
			out.close();
		    } catch (IOException ex) {
			Logger.getLogger(MultiThreadBufferSplitter.class.getName()).log(Level.SEVERE, null, ex);
		    }
		}
		out = null;

		buffer = null;
		wholeFile = null;
		filePiece = null;
	    }
	}
    }
}
